#include "EventLoop.h"
#include "Channel.h"
#include "Logger.h"
#include "Poller.h"
#include <sys/eventfd.h>
#include <unistd.h>

// 防止一个线程创建多个loop
static thread_local EventLoop *t_loop_in_this_loop = nullptr;
// 定义默认超时时间
const int kPollTimeMs = 10000; // ms

// 创建eventfd，用来唤醒subLoop
int create_eventfd() {
    int evtfd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0) {
        LOG_FATAL("eventfd error, errno:%d", errno);
    }
    return evtfd;
}

EventLoop::EventLoop()
    : looping_(false), quit_(false), callingPendingFunctors_(false),
      threadId_(CurrentThread::tid()), poller_(Poller::newDefaultPoller(this)),
      wakeupFd_(create_eventfd()),
      wakeupChannel_(new Channel(this, wakeupFd_)) {
    LOG_DEBUG("EventLoop created %p in "
              "thread %d",
              this, threadId_);
    if (t_loop_in_this_loop) {
        LOG_FATAL("Another EventLoop %p "
                  "exists in this thread %d",
                  t_loop_in_this_loop, threadId_);
    } else {
        t_loop_in_this_loop = this;
    }

    // 设置wakeupFd_感兴趣的事件及相应的回调操作
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
    // 每个loop都将监听wakeupChannel_的EPOLLIN事件
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop() {
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    close(wakeupFd_);
    t_loop_in_this_loop = nullptr;
}

void EventLoop::handleRead() {
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof one);
    if (n != sizeof one) {
        LOG_ERROR("read %d bytes "
                  "instead of %d",
                  n, sizeof one);
    }
}

void EventLoop::loop() {
    looping_ = true;
    quit_ = false;
    LOG_INFO("EventLoop %p start looping", this);

    while (!quit_) {
        activeChannels_.clear();
        // 监听两类fd，clientfd和wakeupfd
        pollReturnTime_ = poller_->poll(
            kPollTimeMs,
            &activeChannels_); // poller监听哪些channel发生了事件，上报给eventloop
        for (Channel *channel : activeChannels_) {
            // eventloop通知channel处理相应的事件
            channel->handleEvent(pollReturnTime_);
        }
        // 执行当前EventLoop事件循环需要处理的回调操作
        // mainLoop会先为subLoop注册回调，然后再唤醒subLoop
        doPendingFunctors();
    }
    LOG_INFO("EventLoop %p stop looping", this);
    looping_ = false;
}

void EventLoop::quit() {
    quit_ = true;
    if (!isInLoopThread()) {
        wakeup();
    }
}

void EventLoop::runInLoop(Functor cb) {
    if (isInLoopThread()) {
        cb();
    } else {
        // 在非当前loop线程中执行cb，就需要唤醒loop所在线程，然后执行cb
        queueInLoop(cb);
    }
}

void EventLoop::queueInLoop(Functor cb) {
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }
    // 唤醒上面需要执行回调的线程
    if (!isInLoopThread() || callingPendingFunctors_) {
        wakeup();
    }
}

// 向wakeupFd_写一个数据来达到唤醒的目的，wakupChannel发生读事件，loop就会唤醒
void EventLoop::wakeup() {
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof one);
    if (n != sizeof one) {
        LOG_ERROR("write %d bytes "
                  "instead of %d",
                  n, sizeof one);
    }
}

void EventLoop::updateChannel(Channel *channel) {
    poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel *channel) {
    poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel *channel) {
    return poller_->hasChannel(channel);
}

void EventLoop::doPendingFunctors() {
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;
    {
        std::unique_lock<std::mutex> lock(mutex_);
        // 把pendingFunctors_的内容转换到局部，解放pendingFunctors_，而且局部对象执行完成后会由栈销毁
        functors.swap(pendingFunctors_);
    }
    for (const Functor &functor : functors) {
        functor(); // 执行当前loop需要执行的回调操作
    }
    callingPendingFunctors_ = false;
}
